package com.bml.architect.springboot;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
public class ConsumerListener {


    @RabbitListener(queues = "springboot_queue")
    public void simpleListener(String message) {

        System.out.println("接受到的消息 ：" + message);


    }

    /**
     * confirmListener监听的是spring_queue_confirm队列，这个队列绑定在交换机spring_confirm_exchange
     * 上，并且交换机的类型是direct
     * <p>
     * 当消费端启动监听的时候，如果说mq中还没有队列与交换机的绑定关系，那么说服务端还没有启动创建对应的交换机和
     * 队列，这个时候如果要启动成功就必须设置bindings参数来绑定交换机和队列的关系
     * 如果说服务提供方已经启动创建了队列和交换机，并且建立了绑定关系，那么就可以直接使用
     *
     * @param message
     * @param channel
     * @RabbitListener(queues = "spring_queue_confirm")来监听队列，说白了就是
     * bindings参数消费端也可以自动创建交换机和队列，并且建立绑定关系
     */
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "spring_confirm_exchange", durable = "true", type = "direct"),
            value = @Queue(value = "spring_queue_confirm", durable = "true"),
            key = "confirm"
    ))
    // @RabbitListener(queues = "spring_queue_confirm")
    @RabbitHandler
    public void confirmListener(Message message, Channel channel) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        String msg = new String(message.getBody());
        System.out.println("接收到的确认消息 ：" + msg);
        try {
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /********************************发布订阅 start*********************************************/
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "fanout_exchange", durable = "true", type = "fanout"),
            value = @Queue(value = "fanout_queue_1", durable = "true")
    ))
    @RabbitHandler
    public void fanoutQ1Listener(Message message, Channel channel) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String msg = new String(message.getBody());
        System.out.println("订阅获取消息Q2：" + msg);
        try {
            channel.basicAck(deliveryTag, true);
        } catch (IOException e) {
            try {
                //第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会重新发送该消息给消费端
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }

    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "fanout_exchange", durable = "true", type = "fanout"),
            value = @Queue(value = "fanout_queue_2", durable = "true")
    ))
    @RabbitHandler
    public void fanoutQ2Listener(Message message, Channel channel) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String msg = new String(message.getBody());
        System.out.println("订阅获取消息Q1：" + msg);
        try {
            channel.basicAck(deliveryTag, true);
        } catch (IOException e) {
            try {
                //第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会重新发送该消息给消费端
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }
    /********************************发布订阅 end*********************************************/
    /********************************路由模式 start*********************************************/

    @RabbitListeners({@RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "direct_exchange", durable = "true", type = "direct"),
            value = @Queue(value = "direct_queue_1", durable = "true"),
            key = "d1.1"
    )), @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "direct_exchange", durable = "true", type = "direct"),
            value = @Queue(value = "direct_queue_2", durable = "true"),
            key = "d2.1"
    ))})
    @RabbitHandler
    public void directQ1Listener(Message message, Channel channel) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String msg = new String(message.getBody());
        System.out.println("获取direct消息：" + msg);
        try {
            channel.basicAck(deliveryTag, true);
        } catch (IOException e) {
            try {
                //第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会重新发送该消息给消费端
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }
    /********************************路由模式 end*********************************************/
    /********************************topics模式 START*********************************************/

    @RabbitListeners({
            @RabbitListener(bindings = @QueueBinding(
                    exchange = @Exchange(value = "topic_exchange", durable = "true", type = "topic"),
                    value = @Queue(value = "topic_queue_1", durable = "true"),
                    key = "t1.#"
            )),
            @RabbitListener(bindings = @QueueBinding(
                    exchange = @Exchange(value = "topic_exchange", durable = "true", type = "topic"),
                    value = @Queue(value = "topic_queue_1", durable = "true"),
                    key = "t2.#"
            ))
    })
    @RabbitHandler
    public void topicListener(Message message, Channel channel) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String msg = new String(message.getBody());
        System.out.println("获取topic消息：" + msg);
        try {
            channel.basicAck(deliveryTag, true);
        } catch (IOException e) {
            try {
                //第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会重新发送该消息给消费端
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }
    /********************************topics模式 end*********************************************/


    /**
     * 延迟队列，监控延迟队列的消息
     */
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "delay_dlx_exchange",durable = "true",type = "topic"),
            value = @Queue(value = "delay_dlx_queue",durable = "true"),
            key="dlx.#"
    ))
    @RabbitHandler
    public void delayListener(Message message,Channel channel){
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String msg = new String(message.getBody());
        System.out.println("接受到的延迟消息：" + msg);
        try {
            channel.basicAck(deliveryTag, true);
        } catch (IOException e) {
            try {
                //第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会重新发送该消息给消费端
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }
}
